package com.atguigu.realtime.app

import com.atguigu.gmall.common.Constant
import com.atguigu.realtime.bean.OrderInfo
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.json4s.CustomSerializer
import org.json4s.JsonAST.{JDouble, JString}
import org.json4s.jackson.JsonMethods

/**
 * Author atguigu
 * Date 2020/10/19 9:52
 */
object OrderApp extends BaseApp {
    override val appName: String = "OrderApp"
    override val master: String = "local[2]"
    override val batchTime: Int = 3
    override val groupId: String = "OrderApp"
    override val topic: String = Constant.ORDER_INFO_TOPIC
    
    /*object String2Double extends CustomSerializer[Double](format => (
        {
            case JString(s) => s.toDouble
        },
        {
            case d: Double => JDouble(d)
        }
    ))*/
    val String2Double = new  CustomSerializer[Double](format => (
        {
            case JString(s) => s.toDouble
        },
        {
            case d: Double => JDouble(d)
        }
    ))
    
    override def run(ssc: StreamingContext, sourceStream: DStream[String]): Unit = {
        //        val orderInfoStream = sourceStream.map(json => JSON.parseObject(json, classOf[OrderInfo]))
        
        val orderInfoStream = sourceStream.map(json => {
            println(json)
            implicit val f = org.json4s.DefaultFormats + String2Double
            JsonMethods.parse(json).extract[OrderInfo]
        })
    
        orderInfoStream.print()
        
    }
    
    
}

/*
json4s  => json for scala
 */